package com.heima.schedule.service.impl;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.service.TaskService;
import com.heima.schedule.service.TaskinfoLogsService;
import com.heima.schedule.service.TaskinfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author 霍森彪
 * @date 2022/7/18 17:36
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoService taskinfoService;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private CacheService cacheService;


    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public long addTask(Task task) {
        //判空
        if (task == null) {
            log.warn("任务参数为空");
            throw new RuntimeException("任务为空");
        }

        //插入taskInfo表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        //因为task中的excuteTime为时间戳,taskInfo中为Date类型,所以需要手动转换赋值
        //类型转换
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        boolean result = taskinfoService.save(taskinfo);
        if (!result) {
            log.warn("插入taskInfo表失败");
            throw new RuntimeException("任务添加失败");
        }

        task.setTaskId(taskinfo.getTaskId());
        //插入taskInfoLog表
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, taskinfoLogs);
        taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
        //设置状态为待执行
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        boolean resultLog = taskinfoLogsService.save(taskinfoLogs);
        if (!resultLog) {
            log.warn("插入taskInfoLog表失败");
            throw new RuntimeException("任务日志添加失败");
        }

        //当前任务执行时间
        long executeTime = task.getExecuteTime();
        //系统当前时间
        long nowTime = System.currentTimeMillis();
        //判断任务执行时间是否小于当前时间
        if (executeTime <= nowTime) {
            cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
            return task.getTaskId();
        }

        //判断任务执行时间是否小于未来五分钟
        if (executeTime <= nowTime + 5 * 60 * 1000) {
            cacheService.zAdd("FUTURE", JSON.toJSONString(task), executeTime);
        }
        return task.getTaskId();
    }
    //删除任务
    @Transactional(rollbackFor = RuntimeException.class)
    @Override
    public Boolean deleteTask(Long taskId) {
        //判空
        if (taskId == null) {
            log.warn("参数为空");
            return false;
        }
        //删除task_info
        boolean deleteResult = taskinfoService.removeById(taskId);
        if (!deleteResult) {
            log.warn("删除task_info失败");
            throw new RuntimeException("删除task_info失败");
        }
        //更新task_info_log
        TaskinfoLogs logs = new TaskinfoLogs();
        //查找当前taskinfolog
        LambdaQueryWrapper<TaskinfoLogs> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(TaskinfoLogs::getTaskId, taskId);
        TaskinfoLogs olderTaskInfoLog = taskinfoLogsService.getOne(wrapper);
        //更新状态为已取消
        BeanUtils.copyProperties(olderTaskInfoLog, logs);
        logs.setStatus(ScheduleConstants.CANCELLED);
        boolean updateResult = taskinfoLogsService.updateById(logs);
        if (!updateResult) {
            log.warn("更新task_info_log失败");
            throw new RuntimeException("更新task_info_log失败");
        }
        //从执行队列删除
        //查找到当前task
        Taskinfo task = taskinfoService.getById(taskId);
        cacheService.lRemove("TOPIC", 0, JSON.toJSONString(task));
        //从未来队列删除
        cacheService.zRemove("FUTURE", 0, JSON.toJSONString(task));
        return true;
    }

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Task poll() {
        //1.指定topic队列,进行消费,执行pop弹出方法,返回的是Task结构的json字符串
        String topic = cacheService.lRightPop("TOPIC");
        if(StringUtils.isEmpty(topic)) {
            log.warn("没有任务");
            return null;
        }
        //将topic转换为Task对象
        Task task = JSON.parseObject(topic, Task.class);

        //删除taskInfo表记录
        boolean removeTaskInfoResult = taskinfoService.removeById(task);
        if (!removeTaskInfoResult) {
            log.warn("删除taskInfo表记录失败");
            return null;
        }

        //更新log表记录
        TaskinfoLogs newTaskInfoLog = new TaskinfoLogs();
        BeanUtils.copyProperties(task, newTaskInfoLog);
        newTaskInfoLog.setStatus(ScheduleConstants.EXECUTED);
        boolean updateResult = taskinfoLogsService.updateById(newTaskInfoLog);
        if (!updateResult) {
            log.warn("更新log表失败");
           return null;
        }
        return task;
    }

    //定时刷新任务
    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void refresh() {
        String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if (StringUtils.isBlank(token)) {
            return;
        }
        System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");

        //获取该组key下需要消费的任务数据
        //获取0~当前时间的所有任务(代表小于当前时间的任务)
        //查询未来队列,获取所有任务
        Set<String> tasks = cacheService.zRangeByScore(ScheduleConstants.FUTURE, 0, System.currentTimeMillis());
        if (CollectionUtils.isEmpty(tasks)) {
            log.warn("任务队列为空");
            return;
        }
        //否则的话,将这些任务添加到消费者队列中
        cacheService.refreshWithPipeline(ScheduleConstants.FUTURE, ScheduleConstants.TOPIC, tasks);
        System.out.println("成功的将" + ScheduleConstants.FUTURE + "下的当前需要执行的任务数据刷新到" + ScheduleConstants.TOPIC + "下");
    }

    //初始化队列数据
    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void reloadData() {
        //清除redis中的数据
        clear();
        //查询mysql中的taskinfo表,读取里面的任务
        List<Taskinfo> taskinfoList = taskinfoService.list();
        if (CollectionUtils.isEmpty(taskinfoList)) {
            log.warn("任务为空");
            return;
        }
        for (Taskinfo taskinfo : taskinfoList) {
            if (taskinfo == null) {
                log.warn("任务为空");
                return;
            }
            //任务执行时间
            long executeTime = taskinfo.getExecuteTime().getTime();
            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(executeTime);

            addTask(task);
        }
    }

    //清除redis中的数据
    public void clear() {
        cacheService.delete("TOPIC");
        cacheService.delete("FUTURE");
    }
}
